package org.firebug.spring.boot.kafka.service;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;

@Component
public class KafkaService {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    public void produce(String topic, String data) {
        ListenableFuture future = kafkaTemplate.send(topic, data);
        kafkaTemplate.setProducerListener(new ProducerListener<String, String>() {
            @Override
            public void onSuccess(String topic, Integer partition, String key, String value, RecordMetadata recordMetadata) {
                System.out.println("onSuccess()");
            }

            @Override
            public void onError(String topic, Integer partition, String key, String value, Exception exception) {
                System.out.println("onError()");
            }

            @Override
            public boolean isInterestedInSuccess() {
                System.out.println("isInterestedInSuccess()");
                return true;
            }
        });
    }

    @KafkaListener(group = "access", topics = {"access"})
    public void consume(String msg) {
        System.out.println("消息被消费: " + msg);
    }
}